package com.yangcangclub.mq.component;

import com.alibaba.fastjson.JSON;
import com.yangcangclub.commons.utils.StringUtil;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.yangcangclub.commons.exception.ApolloException;
import com.yangcangclub.commons.result.MessageEnum;
import com.yangcangclub.commons.utils.NetUtil;
import com.yangcangclub.mq.base.MQBaseDomain;
import com.yangcangclub.mq.bean.MQMessageInfo;
import com.yangcangclub.mq.bean.MQOffsetInfo;
import com.yangcangclub.mq.bean.MQTopicInfo;
import com.yangcangclub.mq.domain.MQLogProfile;
import com.yangcangclub.mq.domain.MQMessageProfile;
import com.yangcangclub.mq.domain.MQOffsetProfile;
import com.yangcangclub.mq.domain.MQTopicProfile;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;

@Component
public class MqComponent extends MQBaseDomain {

    private static org.slf4j.Logger logger = LoggerFactory.getLogger(MqComponent.class);
//    private static Producer producer;

    @Autowired
    private MQTopicProfile mqTopicProfile;

    @Autowired
    private MQMessageProfile mqMessageProfile;

    @Autowired
    private MQOffsetProfile mqOffsetProfile;

    @Autowired
    private MQLogProfile mqLogProfile;


//    @PostConstruct
//    public void init(){
//        if(producer==null&& StringUtil.isEquals("mq", sysConfig.getMqIsOpenModel())) {
//            Properties properties = new Properties();
//            properties.put(PropertyKeyConst.ProducerId, sysConfig.getCommunity_relation_productor());
//            properties.put(PropertyKeyConst.AccessKey, sysConfig.getMqAccessKey());
//            properties.put(PropertyKeyConst.SecretKey, sysConfig.getMqSecretKey());
//            producer = ONSFactory.createProducer(properties);
//            producer.start();
//        }
//    }

    public void sendMqMsg(String topic,String tag,String key,Map<String,Object> map){
        String content= JSON.toJSONString(map);
        if(StringUtil.isEquals("mq",sysConfig.getMqIsOpenModel())) {
//            try {
//                Message msg = new Message(topic, tag, content.getBytes());
//                msg.setKey(key);
//                producer.send(msg);
//            } catch (Exception e) {
//                logger.error("send community mq fail topic={},tag={},key={},content={}", new Object[]{topic,tag, key, content});
//                throw new ApolloException(MessageEnum.MQ_SEND_FAIL);
//            }
        } else if(StringUtil.isEquals("card",sysConfig.getMqIsOpenModel())) {
            this.addMQTopicMessage(topic, key, tag, content);
        }
    }

    /**
     * 添加消息
     * @param topic
     * @return
     */
    public boolean addMQTopicMessage(String topic, String key, String tag, String content){
        String serverIp = NetUtil.getServerIpAddress();
        if(null==serverIp) {
            throw new ApolloException(MessageEnum.GET_SERVER_IP_ERROR);
        }

        MQTopicInfo mqTopicInfo = mqTopicProfile.getServerMQTopic(serverIp,topic);
        if(null!=mqTopicInfo) {
            mqMessageProfile.addMQTopicMessage(mqTopicInfo.getPartition_(),topic,key,tag,content);
            return true;
        }

        return false;
    }

    /**
     * 按tag获取服务器所有未消费的消息
     * @return
     */
    public List<Map<String,Object>> getServerMQTopicMessageList(String topic,String tag,int offset){
        List<Map<String,Object>> data = Lists.newArrayList();

        String serverIp = NetUtil.getServerIpAddress();
        if(null==serverIp) {
            throw new ApolloException(MessageEnum.GET_SERVER_IP_ERROR);
        }

        MQTopicInfo mqTopicInfo = mqTopicProfile.getServerMQTopic(serverIp, topic);
        if(null!=mqTopicInfo) {
            List<MQMessageInfo> mqMessageInfoList = mqMessageProfile.getServerMQTopicMessageList(topic,mqTopicInfo.getPartition_(),tag,offset);
            if(null!=mqMessageInfoList&&mqMessageInfoList.size()>0) {
                for (MQMessageInfo mqMessageInfo:mqMessageInfoList) {
                    Map<String,Object> map = Maps.newHashMap();
                    map.put("messageId",mqMessageInfo.getId());
                    map.put("messagePartition",mqMessageInfo.getPartition_());
                    map.put("messageTopic",mqMessageInfo.getTopic());
                    map.put("messageKey",mqMessageInfo.getKey_());
                    map.put("messageTag",mqMessageInfo.getTag());
                    map.put("messageIsConsume",mqMessageInfo.getIs_consume());
                    map.put("messageContent",mqMessageInfo.getContent());
                    data.add(map);
                }
            }
        }

        return  data;
    }

    /**
     * 获取offset
     * @param consumer
     * @param topic
     * @param tag
     * @return
     */
    public Map<String,Object> getMQOffset(String consumer,String topic, String tag){
        Map<String,Object> map = Maps.newHashMap();
        String serverIp = NetUtil.getServerIpAddress();
        if(null==serverIp) {
            throw new ApolloException(MessageEnum.GET_SERVER_IP_ERROR);
        }
        MQTopicInfo mqTopicInfo = mqTopicProfile.getServerMQTopic(serverIp, topic);
        if(null!=mqTopicInfo) {
            MQOffsetInfo mqOffsetInfo = mqOffsetProfile.getMQOffset(consumer,mqTopicInfo.getPartition_(),topic,tag);
            map.put("id",mqOffsetInfo.getId());
            map.put("offset",mqOffsetInfo.getOffset());
        }

        return map;
    }

    /**
     * 更新消息offset
     * @param id
     * @return
     */
    public boolean updateMqOffset(int id,int offset) {
        return mqOffsetProfile.updateMqOffset(id,offset);
    }

    /**
     * 保存日志
     * @param consumer
     * @param topic
     * @param tag
     * @param message_id
     * @return
     */
    public int addMQLog(String consumer,String topic, String tag,int message_id){
        String serverIp = NetUtil.getServerIpAddress();
        if(null==serverIp) {
            throw new ApolloException(MessageEnum.GET_SERVER_IP_ERROR);
        }
        MQTopicInfo mqTopicInfo = mqTopicProfile.getServerMQTopic(serverIp, topic);
        if(null!=mqTopicInfo) {
            int id = mqLogProfile.addMQLog(consumer,mqTopicInfo.getPartition_(),topic, tag,message_id,0);
            return id;
        }

        return 0;
    }

    /**
     * 更新日志状态
     * @param id
     * @param state
     * @return
     */
    public boolean updateMqLog(int id,int state) {
        return mqLogProfile.updateMqLog(id,state);
    }
}
